package com.itcast.flink.screen.database;

import com.itcast.flink.screen.database.bo.GoodsAccessLog;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.springframework.boot.Banner;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.transaction.annotation.EnableTransactionManagement;

import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;

/**
 * <p>Description: </p>
 *
 * @author
 * @version 1.0
 * <p>Copyright:Copyright(c)2020</p>
 * @date
 */
@SpringBootApplication
@ComponentScan(basePackages = {"com.itcast"})
@EnableTransactionManagement
public class ScreenUniqueVisitorApplication implements CommandLineRunner {
    
    public static void main(String[] args) {
        SpringApplication application = new SpringApplication(ScreenUniqueVisitorApplication.class);
        application.setBannerMode(Banner.Mode.OFF);
        application.run(args);
        
    }
    
    @Override
    public void run(String... args) throws Exception {
        // 运行Flink任务
        executeFlinkTask();
    }
    
    /**
     * 执行flink任务处理
     *
     * @throws Exception
     */
    public void executeFlinkTask() throws Exception {
        // 1. 创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
        env.setParallelism(1);
        
        // 2. 读取Socket数据源
//        DataStreamSource<String> socketStr = env.socketTextStream("192.168.23.128", 9911, "\n");
        DataStreamSource<String> socketStr = env.readTextFile("./data/goods_access.log");
        
        // 3. 数据解析转换处理
        socketStr.flatMap(new FlatMapFunction<String, GoodsAccessLog>() {
            
            @Override
            public void flatMap(String value, Collector<GoodsAccessLog> out) throws Exception {
                // 获取Json中的data数据
                // 根据分隔符解析数据
                String[] arrValue = value.split("\t");
                System.out.println("receive msg => " + value);
                // 将数据组装为对象
                GoodsAccessLog log = new GoodsAccessLog();
                for (int i = 0; i < arrValue.length; i++) {
                    if (i == 0) {
                        log.setIp(arrValue[i]);
                    } else if (i == 1) {
                        log.setAccessTime(Long.valueOf(arrValue[i]));
                    } else if (i == 2) {
                        log.setEventType(arrValue[i]);
                    } else if (i == 3) {
                        log.setGoodsId(arrValue[i]);
                    }
                }
                out.collect(log);
            }
        })
                .filter(new FilterFunction<GoodsAccessLog>() {
                    @Override
                    public boolean filter(GoodsAccessLog value) throws Exception {
                        return value.getEventType().equals("view");
                    }
                })
                .keyBy(GoodsAccessLog::getGoodsId)
                .timeWindow(Time.seconds(10))
                .process(new ProcessWindowFunction<GoodsAccessLog, Map<String, String>, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context context, Iterable<GoodsAccessLog> elements, Collector<Map<String, String>> out) throws Exception {
                        Set<String> ipSet = new HashSet<>();
                        Map<String, String> goodsUV = new LinkedHashMap<>();
                        elements.forEach(log -> {
                            ipSet.add(log.getIp());
                        });
                        goodsUV.put(key, context.window().getEnd() + ":" + ipSet.size());
                        out.collect(goodsUV);
                    }
                })
                .print("uv result").setParallelism(1);
        
        // 5. 执行任务
        env.execute("job");
    }
}
